package com.zw.kafka;

import com.alibaba.fastjson.JSON;
import com.zw.entity.Order;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author morningstar
 * @date 2021/11/30
 * @desc
 */
public class MyProducer {

    public final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws InterruptedException {

        // 1.设置参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "tajlmt0law.cogiot.net:39092,tajlmt0law.cogiot.net:39093,tajlmt0law.cogiot.net:39094");
        /**
         * （1）acks=0 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下⼀条消息。性能最⾼，但是最容易丢消息。
         * （2）acks=1 ⾄少要等待leader已经成功将数据写⼊本地log，但是不需要等待所有follower是否成功写⼊。就可以继续发送下⼀条消息。
         *            这种情况下，如果follower没有成功备份数据，⽽此时leader⼜挂掉，则消息会丢失。
         * （3）acks=-1或all 需要等待 min.insync.replicas(默认为1，推荐配置⼤于等于2) 这个参数配置的副本个数都成功写⼊⽇志，
         *                  这种策略会保证只要有⼀个备份存活就不会丢失数据。这是最强的数据保证。⼀般除⾮是⾦融级别，或跟钱打交道的场景才会使⽤这种配置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");

        /**
         * 发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，⽐如⽹络抖动，
         * 所以需要在接收者那边做好消息接收的幂等性处理
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 设置重试间隔时间  默认 100ms
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);

        // 设置发送消息的本地缓存区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);

        /**
         * Kafka本地线程会从缓冲区读取数据，批量发送到broker
         * 设置批量发送消息的大小默认值是16384，即16KB
         */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);

        /**
         * 默认值是0，意思就是消息必须立即被发送，这样会影响性能
         * 一般设置10ms左右，这个消息发送完后会进入本地的一个batch，10ms内，不管batch是否满16KB，都会发送出去
         */
        props.put(ProducerConfig.LINGER_MS_CONFIG,10);

        // 把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2.创建⽣产消息的客户端，传⼊参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        // 发送5条消息
        int msgNum = 5;
        final CountDownLatch countDownlatch = new CountDownLatch(msgNum);
        for (int i = 0; i <= 5; ++i) {
            Order order = new Order((long) i, i);
            /**
             * 1.String topic: 主题
             * 2.K key: 决定往哪个分区发送消息 计算公式： hash(key)%partitionNum
             * 3.V value: 具体要发送的消息内容
             */
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));

            // 同步
//            try {
//                // 等待消息发送成功的同步阻塞方法
//                RecordMetadata metadata = producer.send(producerRecord).get();
//                // ====阻塞====
//                System.out.println("同步方式发送消息结果：topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
//
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//                // 记录日志 预警系统+1
//                // 设置时间间隔1s 同步的方式再次发送
//                Thread.sleep(1000);
//                try{
//                    RecordMetadata metadata = producer.send(producerRecord).get();
//                } catch (Exception e1){
//                    e1.printStackTrace();
//                }
//            } catch (ExecutionException e) {
//                e.printStackTrace();
//            }

            // 异步回调方式发送消息
            producer.send(producerRecord, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("发送消息失败" + exception.getStackTrace());
                    }
                    if (metadata != null) {
                        System.out.println("异步方式发送消息结果：topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
                    }
                    countDownlatch.countDown(); // -1 = 4  == 0 时，5个回复全部执行完毕
                }
            });

        }

        // 判断latch是不是0，如果不是0，就等待5s
        countDownlatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }
}
